package com.example.kafka.mock;

import com.example.kafka.dao.MetricOpDao;
import com.example.kafka.metric.HostData;
import com.example.kafka.metric.HostReader;
import com.example.kafka.metric.MetricData;
import com.example.kafka.metric.MetricReader;
import com.example.kafka.pojo.MetricInfoV2Pojo;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Builder;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.BeanUtils;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;

import javax.annotation.Resource;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;

/**
 * @author lingGao
 */
@Component
@Slf4j
//@EnableScheduling
public class MetricDataProduct {

    @Resource
    private MetricReader metricReader;
    @Resource
    private HostReader hostReader;
    @Resource
    private ObjectMapper objectMapper;
    @Resource
    private KafkaProducer<String, String> producer;
    @Resource
    private MetricOpDao metricOpDao;

//    @Scheduled(fixedDelay = 1000 * 60)
    public void mockMetricData() throws Exception {
        List<HostData> hostDataList = hostReader.getHostDataList();
        Map<String, HostData> hostnameRefHost = hostDataList.stream()
                .collect(Collectors.toMap(HostData::getHostname, h -> h));
        List<MetricData> metricDataList = metricReader.getMetricDataList();
        Map<String, MetricData> metricDataMap = metricDataList.stream()
                .collect(Collectors.toMap(MetricData::getMetric, m -> m));
        List<MetricInfoV2Pojo> metricInfos = metricOpDao.getAllMetricInfos();
        int total = 0;
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        for (MetricInfoV2Pojo metricInfoV2Pojo : metricInfos) {
            MetricData templateMetricData = metricDataMap.get(metricInfoV2Pojo.getMetricName());
            MetricData metricData;
            for (HostData hostData : hostDataList) {
                if (metricInfoV2Pojo.getMetricName().endsWith("percent")) {
                    metricData = new MetricData();
                    metricData.setMetric(metricInfoV2Pojo.getMetricName());
                    metricData.setTimestamp(System.currentTimeMillis() / 1000);
                    metricData.setEndpoint(hostData.getHostname());
                    metricData.setComponentName("mock component");

                    Calendar ca = Calendar.getInstance();
                    ca.setTime(new Date());
                    if (hostData.getHostname().substring(6, 7).compareTo("h") < 0) {
                        // 30 天递增模拟
                        double d = ca.get(Calendar.DAY_OF_MONTH) + 56.0 + ThreadLocalRandom.current().nextDouble(90) / 100.0;
                        metricData.setValue(String.format("%.2f", d));
                    } else {
                        // 7 天递增模拟
                        double d = ca.get(Calendar.DAY_OF_WEEK) * 13 + ThreadLocalRandom.current().nextDouble(90) / 10.0;
                        metricData.setValue(String.format("%.2f", d));
                    }


                } else {

                    if (templateMetricData == null) {
                        metricData = new MetricData();
                        metricData.setMetric(metricInfoV2Pojo.getMetricName());
                        metricData.setTimestamp(System.currentTimeMillis() / 1000);
                        metricData.setEndpoint(hostData.getHostname());
                        metricData.setComponentName("mock component");
                        metricData.setValue(ThreadLocalRandom.current().nextInt(11) + "");
                    } else {
                        metricData = new MetricData();
                        BeanUtils.copyProperties(templateMetricData, metricData);
                        boolean parsed = false;
                        try {
                            int value = Integer.parseInt(metricData.getValue());
                            if (value > 0) {
                                metricData.setValue(ThreadLocalRandom.current().nextInt(value * 5) + "");
                            }
                            parsed = true;
                        } catch (Exception ignore) {
                        }
                        if (!parsed) {
                            try {
                                double value = Double.parseDouble(metricData.getValue());
                                if (value > 0) {
                                    metricData.setValue(String.format("%.2f",
                                            ThreadLocalRandom.current().nextDouble(value * 10)));
                                }
                            } catch (Exception ignore) {
                            }
                        }
                        metricData.setEndpoint(hostData.getHostname());
                    }
                }
                producer.send(new ProducerRecord<>("mertic",
                        metricData.getEndpoint(),
                        objectMapper.writeValueAsString(toMetric(metricData, hostnameRefHost))), new MetricCallBack());
                total++;
            }
        }
        stopWatch.stop();
        log.info("send mock metric data to kafka, total: {}, cost {} second", total, stopWatch.getTotalTimeSeconds());
    }

    static class MetricCallBack implements Callback {

        public MetricCallBack() {
        }

        /**
         * A callback method the user can implement to provide asynchronous handling of request completion. This method will
         * be called when the record sent to the server has been acknowledged. Exactly one of the arguments will be
         * non-null.
         *
         * @param metadata  The metadata for the record that was sent (i.e. the partition and offset). Null if an error
         *                  occurred.
         * @param exception The exception thrown during processing of this record. Null if no error occurred.
         */
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (metadata == null) {
                log.error("send metric error", exception);
            }
        }
    }

    private Metric toMetric(MetricData metricData, Map<String, HostData> hostnameRefHost) {
        return Metric.builder()
                .endpoint(metricData.getEndpoint())
                .metric(metricData.getMetric())
                .counterType("COUNTER")
                .tags(createTag(metricData, hostnameRefHost))
                .step(60)
                .timestamp(System.currentTimeMillis() / 1000)
                .value(metricData.getValue())
                .build();
    }

    private String createTag(MetricData metricData, Map<String, HostData> hostnameRefHost) {
        String endpoint = metricData.getEndpoint();
        HostData hostData = hostnameRefHost.get(endpoint);
        String system;
        if (hostData != null) {
            system = hostData.getSystem();
        } else {
            system = "能量管理子系统";
        }
        return "component=" + metricData.getComponentName() +
                ",componentalias=CPU,metricgroup=CPUMonitor,metricgroupalias=CPU监视,alias=空闲率,unit=%,datatype=Float," +
                "system=" + system + ",type=device";
    }

    @Builder
    @Getter
    public static class Metric {
        private String endpoint;
        private String metric;
        private String counterType;
        private String tags;
        private int step;
        private long timestamp;
        private String value;
    }

}
